package cn.brian.modular.sys.listener;

import cn.brian.modular.sys.entity.SysUser;
import cn.brian.modular.sys.mapper.SysUserMapper;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Administrator
 * @version 1.0
 **/
@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg_sys1")
public class SysUserProducerTxmsgListener implements RocketMQLocalTransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();
    @Autowired
    private SysUserMapper sysUserMapper;

    @Autowired
    private ServiceImpl serviceImpl;

    //事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        try {

            SysUser oxx=(SysUser)o;System.out.println("oxxx=="+oxx.toString());
            //解析message，sysUserStr
            String transId = (String)message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);

            //方法是返回旧值（即加1前的原始值）测试发送不成功地时候
            int value = transactionIndex.getAndIncrement();
            System.out.printf("#### executeLocalTransaction is executed, msgTransactionId=%s value=%s %n ",
                    transId,value);
            int status = value % 3;
            localTrans.put(transId, status);
            String messageString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            String sysUserStr = jsonObject.getString("sysUser");
            String dataobject=jsonObject.getString("dataobject");
            System.out.printf(dataobject);
            Object c=Class.forName(dataobject).newInstance();

            //sysUserStr（json）SysUser
            Object sysUser = JSONObject.parseObject(sysUserStr,c.getClass());

            if (status == 0) {
                // Return local transaction with success(commit), in this case,
                // this message will not be checked in checkLocalTransaction()
                System.out.printf("    # COMMIT # Simulating msg %s related local transaction exec succeeded! ### %n", message.getPayload());
                //执行本地事务，
//                sysUserMapper.insert(sysUser);
                serviceImpl.save(sysUser);
                return RocketMQLocalTransactionState.COMMIT;
            }

            if (status == 1) {
                // Return local transaction with failure(rollback) , in this case,
                // this message will not be checked in checkLocalTransaction()
                System.out.printf("    # ROLLBACK # Simulating %s related local transaction exec failed! %n", message.getPayload());

                return RocketMQLocalTransactionState.ROLLBACK;
            }
            if (status == 2) {
                // Return local transaction with failure(rollback) , in this case,
                // this message will not be checked in checkLocalTransaction()
                System.out.printf("    # UNKNOWN # Simulating %s related local transaction exec failed! %n", message.getPayload());
                serviceImpl.save(sysUser);
                return RocketMQLocalTransactionState.UNKNOWN;
            }

            System.out.printf("    # UNKNOW # Simulating  related local transaction exec UNKNOWN! \n");

            //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;



        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }


    }

    //事务状态回查，查询是否已保存数据库
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析message，tx_no

        String transId = (String)message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        String keys = (String)message.getHeaders().get(RocketMQHeaders.KEYS);
        RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
        Integer status = localTrans.get(transId);
        if (null != status) {
            switch (status) {
                case 0:
                    retState = RocketMQLocalTransactionState.UNKNOWN;
                    break;
                case 1:
                    retState = RocketMQLocalTransactionState.COMMIT;
                    break;
                case 2:
                    // retState = RocketMQLocalTransactionState.ROLLBACK;
                    break;
            }
        }
        if(status==2){
           SysUser sysUser= sysUserMapper.selectById(keys);
           if(sysUser!=null){
               return RocketMQLocalTransactionState.COMMIT;
           }
        }
        System.out.printf("------ !!! checkLocalTransaction is executed once," +
                        " msgTransactionId=%s, TransactionState=%s status=%s %n",
                transId, retState, status);
        return retState;
    }
}
